package com.shihan.rabbitmq.producer;

import com.rabbitmq.client.*;
import com.shihan.rabbitmq.utils.ConnUtil;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @Description 发送方确认的 Producer
 * @Author shihan
 * @Date 2021/1/2 10:13
 * @Version 1.0
 */
public class SenderConfirm {

    public static final String EXCHANGE_NAME_DIRECT = "exc.direct.name";

    public static final String QUEUE_NAME = "queue.name";

    public static final String BINDING_KEY = "direct.key";

    public static void main(String[] args) throws URISyntaxException, IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException {
        // 创建连接获取 channel
        Channel channel = ConnUtil.getChannel();

        // 将当前 channel 标识为，发送方确认的 channel 类型
        AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();

        // 声明 Exchange 和 Queue，并进行绑定
        channel.exchangeDeclare(EXCHANGE_NAME_DIRECT, BuiltinExchangeType.DIRECT);
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME_DIRECT,BINDING_KEY);

        // 记录 Broker 接收成功、未接收成功的消息 id 集合
        ConcurrentHashMap<Long, String> allMessage = new ConcurrentHashMap<>();
        ConcurrentHashMap<Long, String> failConfirmMessage = new ConcurrentHashMap<>();

        // 发送端确认 -- 异步回调模式
        publisherConfirm_callBack(channel, failConfirmMessage);

        // 发送消息
        publishMessage(channel, allMessage);

        // 打印全部消息，和接收失败的消息 TODO 因为是异步的，所以打印输出的顺序也会有所不同
        System.out.println("全部消息为：\t" + allMessage);
        System.out.println("接收失败的消息为：\t" + failConfirmMessage);

        // 延迟两秒，保证消息被 Broker 成功确认
        try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }

        // 关闭连接通道
        ConnUtil.closeConn();
    }

    /**
     * 正常发送消息
     * @param channel
     * @param allMessage
     */
    private static void publishMessage(Channel channel, ConcurrentHashMap<Long, String> allMessage) throws IOException {
        String currMsg = "";
        for (int i = 0; i < 15; i++) {
            // 获取将要发送的消息对应的 消息ID （应该就是 deliveryTag）
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            currMsg = "publisherConfirm" + i;
            channel.basicPublish(EXCHANGE_NAME_DIRECT,BINDING_KEY,null,(currMsg).getBytes("utf-8"));
            System.out.println("序号为：" + nextPublishSeqNo + " 的消息已发送，但 Broker 尚未确认接收成功！");
            allMessage.put(nextPublishSeqNo,currMsg);
        }

    }

    /**
     * 异步回调方式的 发送端消息确认模式
     * PS：所有待发送的消息，通过回调确认，可以分为 已成功接收 和 未被成功接收，从而可以对未成功接收的消息进行重发
     * @param channel
     * @param failConfirmMessage
     */
    private static void publisherConfirm_callBack(Channel channel, ConcurrentHashMap<Long, String> failConfirmMessage) {
        // Broker 接收到发送端的消息后，触发的回调操作
        ConfirmCallback confirmCallback = new ConfirmCallback() {
            /**
             * 发送端确认的模式：
             * 1、每一条消息都有自己唯一的标识，就是 deliveryTag；
             * 2、multiple 表示 deliveryTag 小于等于当前 deliveryTag 的消息，是否都已经被 Broker 成功接收到了；
             *  2.1、true 表示都已经被成功接收了；
             *  2.2、false 没有都成功接收到，但是当前的 deliveryTag 的消息被成功接收到了
             * @param deliveryTag
             * @param multiple
             * @throws IOException
             */
            @Override
            public void handle(long deliveryTag, boolean multiple) throws IOException {
                if(multiple) {
                    System.out.println("消息编号小于等于 " + deliveryTag + " 的消息，都已经被确认接收了！");
                } else {
                    System.out.println("消息编号为 " + deliveryTag + " 的消息，被成功确认接收了！");
                }
            }
        };

        // 添加监听器
        // 接收成功后的回调，没有接收成功后的回调；
        // channel.addConfirmListener(ConfirmCallback ackCallBack, ConfirmCallback nackCallBack);
        channel.addConfirmListener(confirmCallback, (deliveryTag, multiple) -> {
            if(multiple) {
                failConfirmMessage.put(deliveryTag, "failBeforeAndOwn");
                System.out.println("消息编号小于等于 " + deliveryTag + " 的消息，都没确认！");
            } else {
                failConfirmMessage.put(deliveryTag,"failOwn");
                System.out.println("消息编号为 " + deliveryTag + " 的消息，没被确认！");
            }
        });

    }

}
